#include "skynet.h"
#include "skynet_server.h"
#include "skynet_imp.h"
#include "skynet_mq.h"
#include "skynet_handle.h"
#include "skynet_module.h"
#include "skynet_timer.h"
#include "skynet_monitor.h"
#include "skynet_socket.h"
#include "skynet_daemon.h"
#include "skynet_harbor.h"

#include <pthread.h>
#include <unistd.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

//线程监视器结构
struct monitor {
	int count; //worker线程数
	struct skynet_monitor ** m; //list, 可存放count个skynet_monitor *指针
	pthread_cond_t cond; //条件变量，timer线程激发，worker线程阻塞在它上面
	pthread_mutex_t mutex; //struct monitor内存块的互斥锁
	int sleep; //当前阻塞在条件变量cond上的worker线程数
	int quit; //定时器模块赋值的，被工作线程用来判断是否要退出的标志，(0)不退出 (1)退出
};

struct worker_parm {
	struct monitor *m; //多个工作线程都享有对monitor的指针
	int id; //第x个工作线程
	int weight; //本工作线程对应的消息处理权重
};

static volatile int SIG = 0;

static void
handle_hup(int signal) {
	if (signal == SIGHUP) {
		SIG = 1;
	}
}

#define CHECK_ABORT if (skynet_context_total()==0) break; //如果当前skynet进程中的服务总数为0, 执行break

static void
create_thread(pthread_t *thread, void *(*start_routine) (void *), void *arg) {
	if (pthread_create(thread, NULL, start_routine, arg)) {
		fprintf(stderr, "Create thread failed");
		exit(1);
	}
}

//如果有线程阻塞等待在monitor的条件变量上，则唤醒其中之一
static void
wakeup(struct monitor *m, int busy) {
	if (m->sleep >= m->count - busy) {
		// signal sleep worker, "spurious wakeup" is harmless
		pthread_cond_signal(&m->cond);
	}
}

static void *
thread_socket(void *p) {
	struct monitor * m = p;
	skynet_initthread(THREAD_SOCKET);
	for (;;) {
		int r = skynet_socket_poll();
		if (r==0)
			break;
		if (r<0) {
			CHECK_ABORT
			continue;
		}
		wakeup(m,0); //所有的worker线程都睡在monitor的条件变量上，才唤醒其中之一
	}
	return NULL;
}

//释放monitor占用的系统资源
static void
free_monitor(struct monitor *m) {
	int i;
	int n = m->count;
	for (i=0;i<n;i++) {
		skynet_monitor_delete(m->m[i]);
	}
	pthread_mutex_destroy(&m->mutex);
	pthread_cond_destroy(&m->cond);
	skynet_free(m->m);
	skynet_free(m);
}

//monitor线程
static void *
thread_monitor(void *p) {
	struct monitor * m = p; //强类型转换
	int i;
	int n = m->count;
	skynet_initthread(THREAD_MONITOR);
	for (;;) {
		CHECK_ABORT
		for (i=0;i<n;i++) {
			skynet_monitor_check(m->m[i]);
		}

		//经常让出CPU
		for (i=0;i<5;i++) {
			CHECK_ABORT
			sleep(1);
		}
	}

	return NULL;
}

// make log file reopen
static void
signal_hup() {
	//向日志服务发送了一条type = PTYPE_SYSTEM的skynet_message
	struct skynet_message smsg;
	smsg.source = 0;
	smsg.session = 0;
	smsg.data = NULL;
	smsg.sz = (size_t)PTYPE_SYSTEM << MESSAGE_TYPE_SHIFT;
	uint32_t logger = skynet_handle_findname("logger");
	if (logger) {
		skynet_context_push(logger, &smsg);
	}
}

static void *
thread_timer(void *p) {
	struct monitor * m = p;
	skynet_initthread(THREAD_TIMER);
	for (;;) {
		skynet_updatetime();
		skynet_socket_updatetime();
		CHECK_ABORT
		wakeup(m,m->count-1); //只要睡眠在m->cond上的worker线程数量大于1，就signal唤醒一个worker线程
		usleep(2500);
		if (SIG) {
			signal_hup();
			SIG = 0;
		}
	}
	// wakeup socket thread
	skynet_socket_exit();
	// wakeup all worker thread
	pthread_mutex_lock(&m->mutex);
	m->quit = 1;
	pthread_cond_broadcast(&m->cond);
	pthread_mutex_unlock(&m->mutex);
	return NULL;
}

static void *
thread_worker(void *p) {
	struct worker_parm *wp = p;
	int id = wp->id;
	int weight = wp->weight;
	struct monitor *m = wp->m;
	struct skynet_monitor *sm = m->m[id];
	skynet_initthread(THREAD_WORKER);
	struct message_queue * q = NULL; //q为该worker线程当前正在处理的actor消息队列
	while (!m->quit) {
		q = skynet_context_message_dispatch(sm, q, weight);

		//skynet_context_message_dispatch()返回的是worker线程在死循环的下一次循环时，要处理的actor消息队列
		//若全局队列中已无actor消息队列，令该工作线程阻塞在monitor的条件变量上，陷入休眠，由timer线程或socket线程唤醒
		if (q == NULL) {
			if (pthread_mutex_lock(&m->mutex) == 0) {

				++ m->sleep;

				// "spurious wakeup" is harmless,
				// because skynet_context_message_dispatch() can be call at any time.
				if (!m->quit)
					pthread_cond_wait(&m->cond, &m->mutex);

				-- m->sleep;

				if (pthread_mutex_unlock(&m->mutex)) {
					fprintf(stderr, "unlock mutex error");
					exit(1);
				}
			}
		}
	}
	return NULL;
}

static void
start(int thread) {
	pthread_t pid[thread+3];

	struct monitor *m = skynet_malloc(sizeof(*m));
	memset(m, 0, sizeof(*m));
	m->count = thread;
	m->sleep = 0;

	//相当于是为每个工作线程都分配了一个skynet_monitor结构
	m->m = skynet_malloc(thread * sizeof(struct skynet_monitor *));
	int i;
	for (i=0;i<thread;i++) {
		m->m[i] = skynet_monitor_new();
	}

	//初始化监视器monitor的互斥锁和条件变量
	if (pthread_mutex_init(&m->mutex, NULL)) {
		fprintf(stderr, "Init mutex error");
		exit(1);
	}
	if (pthread_cond_init(&m->cond, NULL)) {
		fprintf(stderr, "Init cond error");
		exit(1);
	}

	//monitor线程用于检测节点内的消息是否堵住，timer线程运行定时器，socket线程进行网络数据的收发
	create_thread(&pid[0], thread_monitor, m);
	create_thread(&pid[1], thread_timer, m);
	create_thread(&pid[2], thread_socket, m);

	//权重表，可以根据需求更改，我们项目用的一般是8核CPU的机子，会用到前8个元素
	static int weight[] = { 
		-1, -1, -1, -1, 0, 0, 0, 0,
		1, 1, 1, 1, 1, 1, 1, 1, 
		2, 2, 2, 2, 2, 2, 2, 2, 
		3, 3, 3, 3, 3, 3, 3, 3, };
	struct worker_parm wp[thread];
	for (i=0;i<thread;i++) {
		wp[i].m = m;
		wp[i].id = i;
		if (i < sizeof(weight)/sizeof(weight[0])) {
			wp[i].weight = weight[i];
		} else {
			wp[i].weight = 0;
		}
		create_thread(&pid[i+3], thread_worker, &wp[i]);
	}

	for (i=0;i<thread+3;i++) {
		pthread_join(pid[i], NULL);  //pthread_join使一个线程以阻塞的方式等待指定的线程结束, skynet启动后主线程阻塞在这里了
	}

	free_monitor(m);
}

static void
bootstrap(struct skynet_context * logger, const char * cmdline) {
	int sz = strlen(cmdline); //cmdline = "snlua bootstrap"
	char name[sz+1];
	char args[sz+1];
	int arg_pos;
	sscanf(cmdline, "%s", name); //只读取了"snlua"这个部分到name中
	arg_pos = strlen(name);
	if (arg_pos < sz) {
		while(cmdline[arg_pos] == ' ') { //跳过空格字符
			arg_pos++;
		}
		strncpy(args, cmdline + arg_pos, sz); //复制"bootstrap"到args
	} else {
		args[0] = '\0';
	}
	struct skynet_context *ctx = skynet_context_new(name, args);
	if (ctx == NULL) {
		skynet_error(NULL, "Bootstrap error : %s\n", cmdline);
		skynet_context_dispatchall(logger);
		exit(1);
	}
}

void 
skynet_start(struct skynet_config * config) {
	// register SIGHUP for log file reopen
	struct sigaction sa;
	sa.sa_handler = &handle_hup; //SIGHUP信号的处理方式为handle_hup
	sa.sa_flags = SA_RESTART; //如果信号中断了其他系统调用, 系统自动重启这一系统调用
	sigfillset(&sa.sa_mask); //SIGHUP信号的处理期间, 屏蔽其他所有信号
	sigaction(SIGHUP, &sa, NULL);

	//检查配置文件内的 deamon 配置, 这个配置对应一个文件路径, 文件内的记录进程的pid号
	//防止重复启动skynet进程, 第一次启动将自动将进程号写入文件
	//并且使本进程忽略了标准输入，标准输出，标准错误
	if (config->daemon) {
		if (daemon_init(config->daemon)) {
			exit(1);
		}
	}

	//初始化节点模块，用于集群，转发远程节点的消息
	skynet_harbor_init(config->harbor);

	//初始化句柄模块，用于给每个actor分配一个全局(一个skynet集群中)唯一的句柄值。
	skynet_handle_init(config->harbor);

	//初始化全局消息队列, 主要是占用的内存分配, 以及置零赋值.
	//全局消息队列里的每一个元素都是一个子消息队列，其中子消息队列的元素内包含具体的消息
	skynet_mq_init();

	//c编写的服务模块的数据结构初始化, 主要用于加载符合 Skynet 服务模块接口的动态链接库
	//初始化了一个 modules 结构体 M，结构体内包含一个 path 成员，保存着配置文件内的 cpath 配置
	skynet_module_init(config->module_path);
	
	//从这里开始看得好困难啊呜呜呜呜, 痛定思痛, 是自己的知识面太窄, 网上搜了解析之后自己再去理解一遍, 很快就看懂了

	//定时器初始化
	skynet_timer_init();

	//初始化socket模块
	skynet_socket_init();

	//设置G_NODE.profile
	skynet_profile_enable(config->profile);

	//加载日志模块, config->logservice是C日志模块的名字, 默认加载的是 项目根路径/cservice/logger.so (主函数中赋予的默认值)
	//config->logger 是logger.so的逻辑中使用的日志输出文件的相对路径, 为空则将输出到标准输出
	struct skynet_context *ctx = skynet_context_new(config->logservice, config->logger);
	if (ctx == NULL) {
		fprintf(stderr, "Can't launch %s service\n", config->logservice);
		exit(1);
	}

	skynet_handle_namehandle(skynet_context_handle(ctx), "logger");

	//相当于开机引导程序, 启动bootstrap服务(./service/bootstrap.lua), 该lua服务不需要别名，执行完skynet.start便结束，内部会启动一些其他的基础服务
	bootstrap(ctx, config->bootstrap);

	//创建各个线程
	start(config->thread);

	// harbor_exit may call socket send, so it should exit before socket_free
	skynet_harbor_exit();
	skynet_socket_free();

	//需要将之前写入了pid的文件删除
	if (config->daemon) {
		daemon_exit(config->daemon);
	}
}
